package pku;

import java.io.*;
import java.util.*;

public class Consumer
{
    private final List<String> topics = new LinkedList<>();
    private int currentTopic = 0;
    private String queue;
    private final Map<String, BufferedInputStream> topicInput = new HashMap<>();

    private ByteMessage readMessage(BufferedInputStream b) throws IOException
    {
        // Get size of Header;
        byte[] size = new byte[4];

        // End Of File;
        if (b.read(size) == -1)
        {
            b.close();
            return null;
        }

        int headerSize = Util.toInt(size);

        if (headerSize < 0)
        {
            System.out.println(headerSize);
            throw new RuntimeException();
        }

        // Get Header;
        byte[] header = new byte[headerSize];
        b.read(header);

        // Get size of body;
        b.read(size);
        int bodySize = Util.toInt(size);

        // Get body;
        byte[] body = new byte[bodySize];
        b.read(body);

        ByteMessage message = new DefaultMessage(body);
        String h = new String(header);
        message.setHeaders(Util.loadHeader(h.split(",")));
        return message;
    }

    private ByteMessage loadByteMessage(String topic) throws IOException
    {
        if (!new File(Util.DATA_PATH + topic).exists())
        {
            return null;
        }

        // If InputStream of current topic is opened in the first place;
        // Then open it and put it in the map;
        if (!topicInput.containsKey(topic))
        {
            InputStream in = new FileInputStream(Util.DATA_PATH + topic);
            BufferedInputStream b = new BufferedInputStream(in);
            topicInput.put(topic, b);
        }

        // Get inputstream and read one message;
        BufferedInputStream buffer = topicInput.get(topic);

        return readMessage(buffer);
    }

    public void attachQueue(String queueName, Collection<String> t) throws Exception
    {
        if (queue != null)
        {
            throw new Exception("Cannot rebinding");
        }
        queue = queueName;
        topics.addAll(t);
    }

    public ByteMessage poll() throws Exception
    {
        ByteMessage re = loadByteMessage(topics.get(currentTopic));
        if (re != null)
        {
            return re;
        }
        else
        {
            // If re == null means that current topic isnt exist or is done with it;
            // Move to next topic;
            currentTopic++;
            // If All of topics have been read;
            if (currentTopic >= topics.size())
                return null;
            else
                return poll();
        }
    }
}
